package com.studerb.tasks.newsGroups;

import java.io.BufferedReader;
import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

import org.apache.commons.io.IOUtils;
import org.apache.commons.net.io.DotTerminatedMessageReader;
import org.apache.commons.net.nntp.NNTPCommand;
import org.apache.commons.net.nntp.NNTPReply;
import org.springframework.beans.factory.annotation.Autowired;

import com.studerb.service.interfaces.INewsGroupService;
import com.studerb.service.interfaces.ITaskService;
import com.studerb.tasks.AbstractTask;
import com.studerb.util.Constants;

public class ListCommandTask extends AbstractTask<Void, Void> {
    private int loopCount = 0;
    private static final int QUEUE_SIZE = 50;
    @Autowired private INewsGroupService newsGroupService;
    @Autowired private ITaskService taskService;
    private BlockingQueue<String> queue;

    @Override public void init() {
        queue = new ArrayBlockingQueue<String>(QUEUE_SIZE);
        this.title = "ListCommand Task";
        this.determinable = true;
        this.connectionNeeded = true;
    }

    private void checkProgress() {
        setMessage("Downloaded: " + loopCount + " NewsGroups");
        if(loopCount % 1000 == 0){
            int currentProgress = (int)(loopCount / 1000);
            if(currentProgress <= 90){
                setProgress(currentProgress);
            }
        }
    }

    private void doListCommand() throws IOException {
        //downloads all groups using 'list' command and updates current list
        setMessage("Sending List Command");
        int reply = this.nntpConnection.sendCommand(NNTPCommand.LIST);
        logger.debug("Got reply: " + this.nntpConnection.getReplyString());
        if(reply != NNTPReply.LIST_OF_NEWSGROUP_FOLLOWS){
            throw new IOException("Error listing groups: reply: " + nntpConnection.getReplyString());
        }
        setProgress(5);
        setMessage("Received: " + this.getConnectionion().getReplyString());
    }

    @Override protected Void doInBackground() throws Exception {
        doListCommand();
        BufferedReader reader = new BufferedReader(new DotTerminatedMessageReader(nntpConnection.getReader()));
        String tempLine;
        setMessage("Downloading NewsGroups");
        Callable<Void> listCommandProcessor = newsGroupService.getListCommandProcessor(this.queue);
        Future processorFuture;
        try{
            processorFuture = taskService.submit(listCommandProcessor);
            while( !isCancelled() && !processorFuture.isDone()){
                if(failed){
                    throw new Exception("Consumer Thread failed");
                }
                loopCount++;
                tempLine = reader.readLine();
                if(tempLine != null){
                    queue.put(tempLine);
                }
                else{
                    queue.put(Constants.END_OF_QUEUE);
                    break;
                }
                checkProgress();
            }
            logger.info("Downloaded: " + loopCount + " lines");
            setProgress(90);
            
            setMessage("Finished Downloading");
            logger.debug("**************************Getting Processor Future get()");
            processorFuture.get();
            return null;
        }
        finally{
            logger.debug("Closing Reader");
            IOUtils.closeQuietly(reader);
            logger.debug("Closing Queue with poison line");
            queue.put(Constants.END_OF_QUEUE);
        }
    }
}
